package com.iacrqq.sedaf.stage.adaptive;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.iacrqq.sedaf.application.adaptive.ApplicationAdjuster;
import com.iacrqq.sedaf.queue.Queue;
import com.iacrqq.sedaf.queue.QueueRuntimeStatistics;
import com.iacrqq.sedaf.stage.Stage;
import com.iacrqq.sedaf.stage.StageRuntimeStatistics;

/**
 * 
 * @author sihai
 * 
 */
public class StageAdjuster
{
	private static Log logger = LogFactory.getLog(StageAdjuster.class);
	
	private static final double DEFAULT_NEW_WEIGHT = 0.75D;
	
	public static final int DEFULT_HISTORY = 10;
	
	public static final double NO_VALUE = -1D;
	
	private int history = DEFULT_HISTORY;
	
	private long interval = ApplicationAdjuster.DEFAULT_ADJUST_INTERVAL;
	
	/**
	 * 过去1<<n, 1<<(n-1) ... 1<<0 interval时间的TPS, NOPR, FR
	 */
	private StageRuntimeStatistics[] _snapshot_history_;
	private double[] _tps_history_;
	private double[] _nopr_history_;
	private double[] _fr_history_;
	private double[] _event_input_queue_ur_history_;
	private double[] _event_input_queue_awt_history_;
	
	private ThreadAdjustTypeEnum threadAdjustType = ThreadAdjustTypeEnum.THREAD_ADJUST_TYPE_ADD;
	private int threadAdjustNumber = 0;
	
	private long total = 0;
	
	private Stage stage;

	public StageAdjuster(Stage stage, int history, int interval)
	{
		assert(stage != null);
		assert(history >= 1);
		assert(interval >= 1);
		
		this.stage = stage;
		this.history = history + 1;
		this.interval = interval;
		this.total = 0;
		
		_snapshot_history_ = new StageRuntimeStatistics[this.history];
		
		_tps_history_ = new double[this.history];
		
		for(int i = 0; i < this.history; i++)
		{
			_tps_history_[i] = NO_VALUE;
		}
		
		_nopr_history_ = new double[this.history];
		
		for(int i = 0; i < this.history; i++)
		{
			_nopr_history_[i] = NO_VALUE;
		}
		
		_fr_history_ = new double[this.history];
		
		for(int i = 0; i < this.history; i++)
		{
			_fr_history_[i] = NO_VALUE;
		}
		
		_event_input_queue_ur_history_ = new double[this.history];
		for(int i = 0; i < this.history; i++)
		{
			_event_input_queue_ur_history_[i] = NO_VALUE;
		}
		
		_event_input_queue_awt_history_ = new double[this.history];
		for(int i = 0; i < this.history; i++)
		{
			_event_input_queue_awt_history_[i] = NO_VALUE;
		}
		
	}

	/**
	 * 
	 */
	public void adjust()
	{
		_update_history_();
		_decision_();
		_adjust_();
		_dump_history_();
	}
	
	public void _update_history_()
	{
		StageRuntimeStatistics currentSnapshot = stage.getStageRuntimeStatistics();
		QueueRuntimeStatistics curretnQueueSnapshot = ((Queue)stage.getInput()).getQueueRuntimeStatistics();
		
		/*System.out.println(new StringBuilder("Stage:").append(stage.getName()).append("'s Runtime Statistics: ")
				.append(currentSnapshot.toString()));*/
		
		// 建立初始基线
		if(_snapshot_history_[0] == null)
		{
			for(int i = 0; i < history; i++)
			{
				_snapshot_history_[i] = currentSnapshot;
			}
			return;
		}
		
		total++;
	
		int index = _get_index_();
		if(index < 0 || index > history)
		{
			System.out.println(new StringBuilder("Wrong index=").append(index).append(", total=").append(total).toString());
			return;
		}
		
		// for 2<<0 to 2<<index
		for(int i = 0; i <= index; i++)
		{
			if(i == 0 || (total % (1 << i)) == 0)
			{
				// stage
				_tps_history_[i] = (0.0D + currentSnapshot.getTs() - _snapshot_history_[i].getTs()) / ((1 << i) * interval);
				_nopr_history_[i] = (0.0D + currentSnapshot.getNops() - _snapshot_history_[i].getNops()) / ((1 << i) * interval);
				_fr_history_[i] = (0.0D + currentSnapshot.getFulls() - _snapshot_history_[i].getFulls()) / ((1 << i) * interval);
				_snapshot_history_[i] = currentSnapshot;
				
				// event input queue
				_event_input_queue_ur_history_[i] = curretnQueueSnapshot.getUsedRate();
				_event_input_queue_awt_history_[i] = curretnQueueSnapshot.getElementAverageWaitingTime();
			}
		}
	}
	
	private int _get_end_()
	{
		int i = 0;
		for(i = 0; i < history; i++)
		{
			if(_tps_history_[i] == NO_VALUE)
			{
				break;
			}
		}
		
		return i;
	}
	
	private int _get_index_()
	{
		return  (int)(Math.log(total) / Math.log(2)) % history; 
	}
	
	private void _decision_()
	{
		/**
		 * 怎样去决策
		 * 
		 * 考虑如下几点：
		 *      这里的都是考虑的当前interval时间内的统计值，要不要向前看一个呢？先不吧
		 *      
		 * 		NOPR >= TPS			上级Stage速度跟不上，或是自己太快，那就要让出1/2的线程出来
		 * 		NOPR >= TPS / 2 	上级Stage速度跟不上，或是自己太快，那就要让出1个线程出来
		 * 
		 * 		FUR >= TPS			下级Stage速度跟不上，或是自己太快，那就让出1/2线程出来(输出Queue大小需要调整，Queue自己会动态调整)
		 * 		FUR >= TPS /2   	下级级Stage速度跟不上，或是自己太快，那就要让出1个线程出来
		 * 
		 *      stage.input.awt
		 *      stage.input.ur
		 * 
		 * 权衡一下，保守一下，每次释放thread，还是一个一个的来，哈哈
		 * 
		 * Event Input Queue
		 * 
		 * 		TPS预测：
		 * 			怎么预测TPS?
		 */
		this.threadAdjustNumber = 0;
		
		if(_nopr_history_[0] >= _tps_history_[0] || _nopr_history_[0] >= (_tps_history_[0] / 2) || _fr_history_[0] >  _tps_history_[0] ||  _fr_history_[0] >  (_tps_history_[0] / 2))
		{
			this.threadAdjustType = ThreadAdjustTypeEnum.THREAD_ADJUST_TYPE_RELEASE;
			this.threadAdjustNumber -= 1;
		}	
		
		// Stage的Event Input Queue
		if(_event_input_queue_ur_history_[0] <= 0.5D)
		{
			this.threadAdjustType = ThreadAdjustTypeEnum.THREAD_ADJUST_TYPE_ADD;
			this.threadAdjustNumber += 1;
		}
		
		int end = _get_end_();
		
		// AWT 预测
		Trend awtTrend = _predict_(_event_input_queue_awt_history_, 0, end);
		if(awtTrend.percent > 0.25)
		{
			this.threadAdjustNumber += 1;
		}
		
		// TPS 预测
		Trend tpsTrend = _predict_(_tps_history_, 0, end);
		if(awtTrend.percent > 0.5)
		{
			this.threadAdjustNumber += 1;
		}
	}
	
	
	private Trend _predict_(double[] historyData, int index, int end)
	{
		/**
		 * 
		 * 
		 * 
		 * 
		 * 
		 * 
		 * 
		 * 
		 * 
		 * 
		 * 
		 */
		int length = end - index;
		if(length == 0 || length == 1)
		{
			return new Trend(TrendTypeEnum.TREND_TYPE_NONE, 0.0D);
		}
		else if(length == 2)
		{
			TrendTypeEnum type = TrendTypeEnum.TREND_TYPE_NONE;
			double percent = (historyData[index] - historyData[end]) / ((historyData[index] + historyData[end]) / 2);
			
			if(percent < 0.0D)
			{
				type = TrendTypeEnum.TREND_TYPE_DECR;
			}
			else if(percent > 0.0D)
			{
				type = TrendTypeEnum.TREND_TYPE_INCR;
			}
			else
			{
				type = TrendTypeEnum.TREND_TYPE_NONE;
			}
			return new Trend(type, 0.0D);
		}
		else
		{
			int middle = index + (end - index) /2;
			Trend newTrend = _predict_(historyData, index, middle);
			Trend oldTrend = _predict_(historyData, middle + 1, end);
			
			double mergedPercent = newTrend.percent + oldTrend.percent;
			TrendTypeEnum type = TrendTypeEnum.TREND_TYPE_NONE;
			if(mergedPercent < 0.0D)
			{
				type = TrendTypeEnum.TREND_TYPE_DECR;
			}
			else if(mergedPercent > 0.0D)
			{
				type = TrendTypeEnum.TREND_TYPE_INCR;
			}
			else
			{
				type = TrendTypeEnum.TREND_TYPE_NONE;
			}
			return new Trend(type, mergedPercent);
		}
	}
	
	private void _adjust_()
	{
		if(threadAdjustNumber > 0)
		{
			logger.warn(new StringBuilder("Stage:").append(stage.getName()).append(" try to add ").append(threadAdjustNumber).append(" threads").toString());
			stage.addThread(threadAdjustNumber);
		}
		else
		{
			logger.warn(new StringBuilder("Stage:").append(stage.getName()).append(" try to remove ").append(0 - threadAdjustNumber).append(" threads").toString());
			stage.removeThread(0 - threadAdjustNumber);
		}
	}
	
	private void _dump_history_()
	{
		StringBuilder sb = new StringBuilder(stage.getName()).append("'s history: tps=");
		
		for(double value : _tps_history_)
		{
			sb.append(_str_value_(value));
			sb.append(",");
		}
		
		sb.append("nopr=");
		for(double value : _nopr_history_)
		{
			sb.append(_str_value_(value));
			sb.append(",");
		}
		
		int index = 0;
		sb.append("fr=");
		for(double value : _fr_history_)
		{
			sb.append(_str_value_(value));
			
			if(index++ != (history - 1))
			{
				sb.append(",");
			}
		}
		
		System.out.println(sb.toString());
	}
	
	private String _str_value_(double value)
	{
		if(value == NO_VALUE)
		{
			return "NO_VALUE";
		}
		else
		{
			return Double.toString(value);
		}
	}
	
	
	private enum ThreadAdjustTypeEnum
	{
		THREAD_ADJUST_TYPE_ADD(0, "Add thread"),
		THREAD_ADJUST_TYPE_RELEASE(1, "Release thread"),
		THREAD_ADJUST_TYPE_NONE(2, "None");
		
		private int value;
		private double percent;
		private String desc;
		
		private ThreadAdjustTypeEnum(int value, String desc)
		{
			this.value = value;
			this.desc = desc;
		}
	}
	
	private enum TrendTypeEnum
	{
		TREND_TYPE_INCR(0, "Increase"),
		TREND_TYPE_DECR(1, "Decrease"),
		TREND_TYPE_NONE(2, "None");
		
		private int value;
		private String desc;
		
		private TrendTypeEnum(int value, String desc)
		{
			this.value = value;
			this.desc = desc;
		}
	}
	
	
	/**
	 * 
	 * @author sihai
	 *
	 */
	private class Trend
	{
		public TrendTypeEnum type = TrendTypeEnum.TREND_TYPE_NONE;
		public double percent = 0.0D;
		
		public Trend(TrendTypeEnum type, double percent)
		{
			this.type = type;
			this.percent = percent;
		}
	}
	
	public static void main(String[] args)
	{
		int history = 10;
	}
}
